package com.yujiahao.bigdata.streaming

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

object Stream_Source_Queue {
  def main(args: Array[String]): Unit = {
    //TODO SparkStreaming环境
    val  conf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //StreamingContext的构造方法第一个参数是配置文件，第二个参数表示数据采集的周期（微批次）
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    //3.创建RDD队列
    val rddQueue = new mutable.Queue[RDD[String]]()

    //4.创建QueueInputDStream
    val inputStream = ssc.queueStream(rddQueue,oneAtATime = false)


    val wordDS: DStream[String] = inputStream.flatMap(_.split(" "))
    val wordToOne: DStream[(String, Int)] = wordDS.map((_, 1))
    val wordToCountDS: DStream[(String, Int) ] = wordToOne.reduceByKey(_ + _)
   //流式数据过来一定要处理，不然会报错
    wordToCountDS.print()
    //8.循环创建并向RDD队列中放入RDD
    for (i <- 1 to 5) {
      rddQueue += ssc.sparkContext.makeRDD(List("a","b","a"), 10)
      Thread.sleep(2000)
    }


    //启动采集器
    // No output operations registered, so nothing to execute
    ssc.start()
    //Driver等待采集器的结束，否则，当前Driver处于阻塞状态
    ssc.awaitTermination()

  }

}
